package mq.rabbit.demo;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

import mq.rabbit.common.util.RabbitMQUtils;
import org.junit.Test;

/**
 * 广播
 * <p>
 * <br />
 * <p>
 * RabbitMQ消息传递模型中的核心思想是生产者  从不直接将任何消息发送到队列。
 * <p>
 * <br />
 * 在广播模式下，消息发送流程是这样的：
 * <li>可以有多个消费者</li>
 * <li>每个消费者有自己的queue（队列）</li>
 * <li>每个队列都要绑定到Exchange（交换机: 交换机类型有几种可用： direct ， topic ， headers  和 fanout 。）</li>
 * <li>生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定</li>
 * <li>队列的消费者都能拿到消息。实现一条消息被多个消费者消费</li>
 */
public class Fanout {

    @Test
    public void testProvider() throws IOException {
        // 获取连接对象
        Connection connection = RabbitMQUtils.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        // 将通道声明指定的交换机
        // 参数1：交换机名称
        // 参数2：交换机类型 fanout表示广播
        channel.exchangeDeclare("logs", "fanout");

        // 发送消息
        // 参数1：交换机名称
        String message = "fanout type message";
        channel.basicPublish("logs", "", null, message.getBytes());

        // 关闭
        RabbitMQUtils.closeChanelAndConnection(channel, connection);
    }

    @Test
    public void testConsumer() throws IOException, InterruptedException {
        for (int i = 0; i < 3; i++) {
            // 获取连接对象
            Connection connection = RabbitMQUtils.getConnection();

            // 获取通道对象
            Channel channel = connection.createChannel();

            // 通道绑定交换机
            channel.exchangeDeclare("logs", "fanout");

            /**
             * 临时队列
             * 首先，无论何时我们连接到Rabbit，我们都需要一个全新的空队列。
             * 为此，我们可以使用随机名称创建一个队列，或者甚至更好-让服务器为我们选择一个随机队列名称。
             *
             * 其次，一旦我们断开与消费者的联系，队列应该是自动删除。
             *
             * 在Java客户端中，当我们不向提供任何参数时 queueDeclare（）
             * 我们使用生成的名称创建一个非持久的，排他的，自动删除队列：
             */
            String queueName = channel.queueDeclare().getQueue();

            // 绑定交换机和队列
            // 参数1：临时队列名称
            // 参数2：交换机名称
            channel.queueBind(queueName, "logs", "");

            final int num = i;
            // 消费消息
            channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("消费者" + num + " --- " + new String(body));
                }
            });
        }

        TimeUnit.MINUTES.sleep(3);
    }
}
